﻿using Apache.NMS;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

namespace HAArtemis.Entity
{
    public class MqConsumer<TEntity> : BaseMq
        where TEntity : Interface.IMqReceiveMsg, new()
    {
        /// <summary>
        /// Listener方法
        /// </summary>
        public MethodInfo? Method { get; set; }
        /// <summary>
        /// 消费者对象
        /// </summary>
        TEntity Entity { get; set; }
        ~MqConsumer()
        {
            Connection.CloseAsync();
        }
        /// <summary>
        /// 默认参数方便使用者进行测试
        /// </summary>
        /// <param name="ipAddress"></param>
        /// <param name="user"></param>
        /// <param name="password"></param>
        public MqConsumer(string ipAddress = "amqp://a:a@localhost:61626", string user = "admin", string password = "admin", string type = "queue", ushort qty = 4, string msgName = "ha_message")
        {
            Url = ipAddress;
            Factory = new NMSConnectionFactory(Url);
            Connection = Factory.CreateConnection(user, password);
            Connection.Start();
            Entity = new TEntity();
            Method = typeof(TEntity).GetMethod("ReceiveMessage");
            if (type.ToLower() == "topic")
            {
                for (ushort i = 0; i < qty; i++)
                {
                    AddTopicListener(msgName, Method);
                }
            }
            else
            {
                for (ushort i = 0; i < qty; i++)
                {
                    AddQueueListener(msgName, Method);
                }
            }
        }

        /// <summary>
        /// 添加Queue监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddQueueListener(string name = "", MethodInfo? method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;

            if (method == null) method = this.Method;

            ISession session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetQueue(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnTextMessage);
            }
        }
        /// <summary>
        /// 添加Topic监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddTopicListener(string name = "", MethodInfo? method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetTopic(name);
            ////创建持久化消费者
            //IMessageConsumer consumer = session.CreateDurableConsumer(session.GetTopic(name), "router");
            IMessageConsumer consumer = session.CreateConsumer(dest);

            if (method != null)
            {
                consumer.Listener += new MessageListener(OnTextMessage);
            }
        }

        /// <summary>
        /// 添加Queue监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddQueueListenerObj(string name = "", MethodInfo? method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetQueue(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnObjectMessage);
            }
        }
        /// <summary>
        /// 添加Topic监听
        /// </summary>
        /// <param name="name"></param>
        /// <param name="method"></param>
        public void AddTopicListenerObj(string name = "", MethodInfo? method = null)
        {
            if (string.IsNullOrWhiteSpace(name)) name = this.Name;
            if (string.IsNullOrWhiteSpace(name)) return;
            if (method == null) method = this.Method;

            ISession session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
            IDestination dest = session.GetTopic(name);
            IMessageConsumer consumer = session.CreateConsumer(dest);
            if (method != null)
            {
                consumer.Listener += new MessageListener(OnObjectMessage);
            }
        }
        /// <summary>
        /// 消费事件
        /// </summary>
        /// <param name="receivedMsg"></param>
        private void OnTextMessage(IMessage receivedMsg)
        {
            if (receivedMsg != null)
            {
#pragma warning disable CS8602 // 解引用可能出现空引用。
                Method?.Invoke(Entity, new string[] { (receivedMsg as ITextMessage).Text });
#pragma warning restore CS8602 // 解引用可能出现空引用。
            }
            else
            {
                Method?.Invoke(Entity, null);
            }
        }
        /// <summary>
        /// 消费事件
        /// </summary>
        /// <param name="receivedMsg"></param>
        private void OnObjectMessage(IMessage receivedMsg)
        {
            if (receivedMsg != null)
            {
#pragma warning disable CS8602 // 解引用可能出现空引用。
                Method?.Invoke(Entity, new object[] { (receivedMsg as IObjectMessage).Body });
#pragma warning restore CS8602 // 解引用可能出现空引用。
            }
            else
            {
                Method?.Invoke(Entity, null);
            }
        }
        /// <summary>
        /// 注册消费者对象和事件
        /// </summary>
        /// <param name="entity"></param>
        /// <param name="method"></param>
        public void OnMessage(TEntity entity, MethodInfo method)
        {
            this.Entity = entity;
            this.Method = method;
        }
    }
}
